package com.snopy.mq.kafka;

import com.alibaba.fastjson.JSON;
import com.snopy.mq.kafka.intercept.ProducerByInterceptor;
import com.snopy.pojo.Banana;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

/**
 * @author snopy
 * @version 1.0.0
 * @classname InterceptProducer
 * @description
 * @email 77912204@qq.com
 * @date 2022/06/09 17:42
 */
public class InterceptProducer {

    public static void main(String[] args) {
        //1.配置消息生产者参数，
        Properties props = new Properties();
        //1.1消息服务器地址
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.164.101:9092");
        //1.2key和value序列化
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());

        //1.3添加发送者拦截器
        props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, ProducerByInterceptor.class.getName());
        //1.4

        //2.生产者参数构建生产者
        KafkaProducer<String,String> producer = new KafkaProducer<String,String>(props);

        //3.构建消息内容
        Banana banana = new Banana("Pink","Jack",42,"this is native banana!");
        ProducerRecord<String,String> record = new ProducerRecord<String,String>("test",JSON.toJSONString(banana));

        //4.发送消息
        producer.send(record);

        //5.关闭发送者

        producer.close();

    }
}
